package cn.darksoul3.spark.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object WCWithStreaming {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("WCWithStreaming").setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(5))
    ssc.sparkContext.setLogLevel("WARN")
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.1.5", 10008)
    val words: DStream[String] = lines.flatMap(_.split("\\s+"))
    val word = words.map((_, 1))

    val reduced = word.reduceByKey(_+_)
    reduced.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
